package com.rem.flink.flink5Watermark;

import com.rem.flink.flink2Source.ClickSource;
import com.rem.flink.flink2Source.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * WindowedStream .reduce()和.aggregate() 和 全窗口函数结合
 *
 * @author Rem
 * @date 2022-10-11
 */

public class AggregateAndProcessWindowFunctionTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps()
                        .withTimestampAssigner((element, recordTimestamp) -> element.getTimestamp()));

        stream.keyBy(Event::getUrl)
                .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))

                /**
                 * 类型形参：
                 * < IN > – 聚合的值的类型（输入值）
                 * < ACC > – 累加器的类型（中间聚合状态）。
                 * < OUT > – 聚合结果的类型
                 */
                .aggregate(new AggregateFunction<Event, Long, Long>() {

                    /**
                     * 创建累加器
                     * @return
                     */
                    @Override
                    public Long createAccumulator() {
                        return 0L;
                    }

                    /**
                     * 数据每来一次就加一次并返回给累加器
                     * @param event
                     * @param accumulator
                     * @return
                     */
                    @Override
                    public Long add(Event event, Long accumulator) {
                        return accumulator + 1;
                    }

                    /**
                     * 从累加器获取聚合结果
                     * 窗口关闭时，增加聚合函数，将结果发送到下游
                     * @param accumulator
                     * @return
                     */
                    @Override
                    public Long getResult(Long accumulator) {
                        return accumulator;
                    }

                    @Override
                    public Long merge(Long a, Long b) {
                        return null;
                    }

                    /**
                     * 使用用于检索额外信息的上下文在键控（分组）窗口上评估的函数的基本抽象类。
                     * 类型形参：
                     * < IN > – 输入值的类型。
                     * < OUT > – 输出值的类型。
                     * < KEY > – 输入的类型。
                     * < W > – 可以应用此窗口函数的Window类型
                     */
                }, new ProcessWindowFunction<Long, UrlViewCount, String, TimeWindow>() {
                    @Override
                    public void process(String url, ProcessWindowFunction<Long, UrlViewCount, String, TimeWindow>.Context context, Iterable<Long> elements, Collector<UrlViewCount> out) {
                        long start = context.window().getStart();
                        long end = context.window().getEnd();
                        out.collect(new UrlViewCount(url, elements.iterator().next(), start, end));
                    }
                })
                .print();

        env.execute();
    }


}
